package com.mq.mqserver;

import com.mq.common.Consumer;
import com.mq.common.MqException;
import com.mq.mqserver.core.*;
import com.mq.mqserver.datacore.DiskDataCenter;
import com.mq.mqserver.datacore.MemoryDataCenter;
import lombok.Data;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 虚拟主机 连接 交换机 队列 绑定 消息 数据
 * 这里不进行多个虚拟机的设置,分辨不同的交换机,通过重新配置交换机的名字:
 *  交换机: exchangeName = virtualHostName+exchangeName
 *  从新配置队列名字: queueName = virtualHostName+queueName
 *  通过名字不同 这样的方式来表示在不同的虚拟机上
 * 在这个类中 处理抛出的异常
 *
 */
@Data
public class VirtualHost {
    private String virtualHostName;
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();

    //创建锁对象,实现线程安全
    //交换机相关的锁
    public final Object exchangeLocker = new Object();
    //队列相关的锁
    public final Object queueLocker = new Object();

    public Router router = new Router();
    //添加一个消费者管理对象
    private ConsumerManager consumerManager = new ConsumerManager( this);

    //初始化操作
    public VirtualHost(String virtualHostName){
        this.virtualHostName = virtualHostName;

        //对磁盘中的数据进行初始化操作,执行建库建表等操作,对于内存中的数据不需要初始化,
        diskDataCenter.init();
        //对内存中的数据,需要将磁盘上的数据拷贝到内存中
        try{
            memoryDataCenter.recovery(diskDataCenter);
        }catch(Exception e){
            System.out.println("[VirtualHost] 恢复内存数据失败" );
            e.printStackTrace();
        }
    }

    /**交换机
     * 创建交换机,将其存入磁盘和内存的对应位置
     * 当多个线程同时创建和删除交换机时,可能会产生线程安全问题
     * 这里要对线程安全进行管理,对交换机的创建和删除,加同一把锁,
     */
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType,
                           boolean isDurable, boolean autoDelete, Map<String,Object> args){
        exchangeName = virtualHostName+exchangeName;
        try{
            synchronized(exchangeLocker){
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if(existsExchange!=null){
                    System.out.println("[VirtualHost] 交换机已经存在 exchangeName:"+exchangeName);
                    return true;
                }
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setDurable(isDurable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArgs(args);
                //将创建好的交换机存储到磁盘和内存中
                //这里先存储磁盘,后存储内存,因为存如磁盘可能会出现异常,存内存出现异常的可能较小
                // 当出现异常后,若先存入内存,还需要将内存中的信息删除,比较麻烦
                //存入磁盘,当为持久化时,才将其存入磁盘中
                if(isDurable) {
                    diskDataCenter.insertExchange(exchange);
                }
                //存入内存
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost] 交换机创建成功!");
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 交换机创建失败! exchangeName:"+exchangeName);
            e.printStackTrace();
            return false;
        }
    }
    //删除交换机
    //将其从磁盘,内存中删除消息
    public boolean exchangeDelete(String exchangeName){
        exchangeName = virtualHostName+exchangeName;
        try {
            synchronized(exchangeLocker){
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if (existsExchange == null) {
                    throw new MqException("[VirtualHost] 交换机不存在,删除失败! exchangeName:" + exchangeName);
                }
                //删除内存上的交换机:
                memoryDataCenter.deleteExchange(exchangeName);
                //当该交换机为持久的,将其从磁盘上删除:
                if(existsExchange.isDurable()){
                    diskDataCenter.deleteExchange(exchangeName);
                }
                //从内存中删除
                memoryDataCenter.deleteExchange(exchangeName);
                System.out.println("[VirtualHost] 删除交换机成功!");
            }
            return true;
        }catch (MqException e) {
            System.out.println("[VirtualHost] 交换机删除失败!");
            e.printStackTrace();
            return false;
        }
    }

    /**队列
     *针对队列的创建和删除,在多线程下,也可能会出现线程安全问题,
     * 需要针对队列的创建和删除加同一把锁
     */
    //创建队列:创建队列并将其存入到磁盘和内存中                                               是否独占
    public boolean queueDeclare(String queueName,boolean isDurable,boolean autoDelete,
                                boolean exclusive,Map<String,Object> args){
        queueName = virtualHostName+queueName;
        try {
            synchronized(queueLocker){
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if(existsQueue!=null){
                    System.out.println("[VirtualHost] 队列已经存在 queueName:"+queueName);
                    return true;
                }
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(isDurable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArgs(args);
                //存入磁盘
                if(isDurable) {
                    diskDataCenter.insertQueue(queue);
                }
                //存入内存
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 创建队列成功 !");
            }
            return true;
        } catch (IOException e) {
            System.out.println("[VirtualHost] 创建队列失败 queueName:"+queueName);
            e.printStackTrace();
            return false;
        }
    }
    //删除队列:从磁盘和内存中 删除队列
    public boolean queueDelete(String queueName){
        queueName = virtualHostName+queueName;
        try{
            synchronized(queueLocker){
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if(existsQueue==null){
                    throw new MqException("[VirtualHost] 队列不存在,删除队列失败 queueName:"+queueName);
                }
                if(existsQueue.isDurable()){
                    diskDataCenter.deleteQueue(queueName);
                }
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost] 删除队列成功 queueName:"+queueName);
            }
            return true;
        }catch (Exception e) {
            System.out.println("[VirtualHost] 队列删除失败! queueName:" +queueName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 绑定
     */
    //添加绑定 :创建绑定并存入磁盘和内存
    public boolean bindingDeclare(String exchangeName,String queueName,String bindingKey){
        exchangeName = virtualHostName+exchangeName;
        queueName = virtualHostName+queueName;
        try{
            //1.验证绑定是否存在:
            Binding existsBinding = memoryDataCenter.getBinding(exchangeName, queueName);
            if(existsBinding!=null){
                throw new MqException("[VirtualHost] 绑定已存在  bindingkey:"+bindingKey);
            }
            //2.判断绑定是否合规
            if(!router.checkBindingKey(bindingKey)){
                throw new MqException("[VirtualHost] 绑定格式有误 bindingKey:"+bindingKey);
            }
            //3.创建绑定
            Binding binding = new Binding();
            binding.setExchangeName(exchangeName);
            binding.setQueueName(queueName);
            binding.setBindingKey(bindingKey);
            //4.验证绑定的队列和交换机是否存在
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if(queue==null){
                throw new MqException("[VirtualHost] 要绑定的队列不存在 queueName:"+queueName+
                        " ,bindingKey:"+bindingKey);
            }
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if(exchange==null){
                throw new MqException("[VirtualHost] 要绑定的交换机不存在 exchangeName:"+exchangeName+
                        " ,bindingKey:"+bindingKey);
            }
            //5,存入磁盘
            if(queue.isDurable() && exchange.isDurable()){
                diskDataCenter.insertBinding(binding);
            }
            //6.存入内存
            memoryDataCenter.insertBinding(binding);
            System.out.println("[VirtualHost] 创建绑定成功 bindingKey: "+bindingKey);
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 创建绑定失败!" );
            e.printStackTrace();
            return false;
        }
    }
    //删除绑定:从磁盘和内存中删除绑定
    public boolean bindingDelete(String exchangeName,String queueName){
        exchangeName = virtualHostName+exchangeName;
        queueName = virtualHostName+queueName;
        try{
            //1.查看绑定是否存在:
            Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
            if(binding==null){
                throw new MqException("[VirtualHost] 绑定不存在 exchangeName:"+exchangeName+
                        " ,queueName:"+queueName);
            }
            //2.从磁盘删除
            //这里本来是要先判断绑定的交换机和队列是否为持久化的,若不是持久化的是不存在磁盘中的,
            //但判断起来比较麻烦,并且要是没在磁盘中存储,删除失败,也不会产生啥影响,因此这里就直接删除了
            diskDataCenter.deleteBinding(binding.getBindingKey());
            //3.从内存删除绑定
            memoryDataCenter.deleteBinding(binding);
            System.out.println("[VirtualHost] 删除绑定成功 exchangeName:"+exchangeName+
                    " ,queueName:"+queueName +" , bindingKey:"+binding.getBindingKey());
            return true;
        }catch(Exception e){
            System.out.println("[VirtualHost] 删除绑定失败 exchangeName:"+exchangeName+
                    " ,queueName:"+queueName);
            e.printStackTrace();
            return false;
        }
    }

    //发送消息到指定的队列
//     public boolean basicPublish1(String exchangeName,String routingKey,BasicProperties basicProperties,byte[] body){
//        exchangeName = virtualHostName+exchangeName;
//        try{
//            //1.先判定交换机是否存在
//            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
//            if(exchange==null){
//                throw new MqException("[VirtualHost] 交换机不存在 exchangeName:"+exchangeName);
//            }
//            //2.判断交换机的类型
//            //直接交换机:专属红包,和bindingKey没有关系
//            //routingKey当成queueName,发送到指定队列
//            if(exchange.getType()==ExchangeType.DIRECT){
//                //判断routingKey格式是否正确
//                if(!router.checkRoutingKey(routingKey)){
//                    throw new MqException("[VirtualHost] routingKey格式有误! routingKey:"+routingKey);
//                }
//                //routingKey就是queueName
//                String queueName = virtualHostName+routingKey;
//                //判断队列是否存在
//                MSGQueue queue = memoryDataCenter.getQueue(queueName);
//                if(queue==null){
//                    throw new MqException("[VirtualHost] 队列不存在 queueName:"+queueName);
//                }
//                //构造消息对象
//                Message message = Message.createMessageWithId(routingKey,basicProperties,body);
//                //发送消息到队列
//                sendMessage(queue,message);
//                System.out.println("[VirtualHost] 消息发送到队列成功");
//            }else{
//                //当交换机的类型为fanout/topic时
//                //fanout 发送给绑定该交换机的所有队列
//                //topic根据routingKey和bindingKey的匹配进行发送消息
//                //遍历所有连接该交换机的所有绑定关系
//                ConcurrentHashMap<String, Binding> bindings = memoryDataCenter.getBindings(exchangeName);
//                for(Map.Entry<String,Binding> bindingMap:bindings.entrySet()){
//                    //这个集合是仅仅将绑定的关系存放了进去,但真实的交换机和队列是否存在,是不确定的
//                    /*
//                    Binding binding = bindingMap.getValue();
//                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
//                     */
//                    String queueName = bindingMap.getKey();
//                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
//                    if(queue==null){
//                        throw new MqException("[VirtualHost] 队列不存在! queueName:"+queueName);
//                    }
//                    Binding binding = bindingMap.getValue();
//                    //当前绑定的交换机和队列都存在了
//                    //构造消息
//                    if(!router.route(exchange.getType(),binding,routingKey)){
//                        throw new MqException("[VirtualHost] 路由不匹配 bindingKey:"+binding.getBindingKey()+
//                                " , routingKey:"+routingKey);
//                    }
//                    //构造消息
//                    Message message = Message.createMessageWithId(routingKey,basicProperties,body);
//                    //发送消息
//                    sendMessage(queue,message);
//                    System.out.println("[VirtualHost] 消息发送成功! queueName:"+queueName +
//                            " ,messageId:"+message.getMessageId());
//                }
//            }
//            return true;
//        }catch (Exception e){
//            System.out.println("[VirtualHost] 消息发送失败 exchangeName:"+exchangeName+
//                    " ,routingKey:"+routingKey);
//            e.printStackTrace();
//            return false;
//
//        }
//    }

     // 消息
    //发送消息到指定的队列
    //根据交换机找到对应的队列,根据路由规则,进行匹配
    public boolean basicPublish(String exchangeName,String routingKey,
                                BasicProperties basicProperties,byte[] body){
        exchangeName = virtualHostName+exchangeName;
        try{
            //1.判定交换机是否存在
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if(exchange==null){
                throw new MqException("[VirtualHost] 交换机不存在 exchangeName: "+exchangeName);
            }
            //2.判断路由格式是否正确
            if(!router.checkRoutingKey(routingKey)){
                throw new MqException("[VirtualHost] 绑定格式有误! routingKey:"+routingKey);
            }
            //3.根据路由类型,进行发送消息
            //direct:直接路由:不使用bindingKey
            //直接交换机:专属红包,和bindingKey没有关系
            //routingKey当成queueName,发送到指定队列
            if(exchange.getType()==ExchangeType.DIRECT){
                //以routingKey作为队列的名字,直接把消息写入指定的队列(专属一个队列)
                String queueName = virtualHostName+routingKey;
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if(queue==null){
                    throw new MqException("[VirtualHost] 队列不存在 queueName:"+queueName);
                }
                //构造消息对象:
                Message message =Message.createMessageWithId(routingKey,basicProperties,body);
               //发送消息到队列中
                sendMessage(queue,message);
                System.out.println("[VirtualHost] 消息发送成功 queueName:"+queueName+
                        " ,messageId:"+message.getMessageId());
            }else{
                //当交换机的类型为fanout/topic时
                //fanout:将消息交给所有绑定交换机的所有队列
                //topic:将消息发送给指定交换机的队列中,要满足routingKey和binding相匹配的队列
                ConcurrentHashMap<String, Binding> bindingMap = memoryDataCenter.getBindings(exchangeName);
                //遍历该交换机所有绑定的 队列和绑定集合
                for(Map.Entry<String,Binding> b : bindingMap.entrySet()){
                    Binding binding = b.getValue();
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if(queue==null){
                        System.out.println("[VirtualHost] 该队列不存在 queueName:"+binding.getQueueName());
                        continue;
                    }
                    //构造消息对象
                    Message message = Message.createMessageWithId(routingKey,basicProperties,body);
                    //判断该交换机时什么类型fanout/topic
                    //fanout:所有人都能领到红包
                    //topic:收红包要说出对应的口令
                    //判断bindingKey和routingKey是否满足匹配:
                    if(!router.route(exchange.getType(),binding,routingKey)) continue;
                    //匹配成功时,就将消息转发
                    sendMessage(queue,message);
                    System.out.println("[VirtualHost] 消息发送成功 queueName:"+queue.getName()+
                            " ,messageId:"+message.getMessageId());
                }
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost]消息发送失败 messageId:");
            e.printStackTrace();
            return false;
        }
    }

    //发送消息
    //将消息写入磁盘和内存中
    private void sendMessage(MSGQueue queue, Message message) {
        try{
            //判断当前消息是否是持久化的
            //  1:非持久化 0:持久化
            int n = message.getDeliveryMode();
            if(n==0){
                //消息为持久化时,将消息写入磁盘
                diskDataCenter.sendMessage(queue,message);
            }
            //将消息写入内存
            memoryDataCenter.sendMessageToQueue(queue,message);
            // 然后就是通知消费者消费消息
            consumerManager.notifyConsumer(queue.getName());
        }catch (Exception e){
            System.out.println("[VirtualHost] 消息发送失败 queueName:"+queue.getName()+
                    " , messageId:"+message.getMessageId());
            e.printStackTrace();
        }
    }


    /**
    //订阅消息
    // 当消息到达队列后,就要将消息推送给对应的订阅者(消费者)
     * @param consumerTag 消费者的身份标识
     * @param queueName 要订阅的队列名字
     * @param autoAck 是否自动确认消息
     * @param consumer 回调函数,是一个函数式接口, 实际执行消息的逻辑代码,
     *                 后续调用basicConsumer方法的时候,填入对应的实参,即可完成自己的代码功能(通过lambda表达式)
     * @return
     */
    public boolean basicConsumer(String consumerTag, String queueName,
                                 boolean autoAck, Consumer consumer){
        //构造一个ConsumerEnv对象,找到要订阅的队列,将consumer方法传入队列
        queueName = virtualHostName+queueName;
        try {
            consumerManager.addConsumer(consumerTag,queueName,autoAck,consumer);
            System.out.println("[VirtualHost] basicConsumer 成功 queueName:"+queueName);
            return true;
        } catch (MqException e) {
            System.out.println("[VirtualHost] basicConsumer 失败 queueName:"+queueName);
            e.printStackTrace();
            return false;
        }
    }

//  手动确认消息模式下要调用该方法,处理消费后的消息
    //删除消息信息
    public boolean basicAck(String queueName,String messageId){
        try {
            //1.获取消息
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if(queue==null){
                throw new MqException("[VirtualHost] 队列不存在 queueName:"+queueName);
            }
            //2.获取队列
            Message message = memoryDataCenter.getMessage(messageId);
            if(message==null){
                throw new MqException("[VirtualHost] 消息不存在 messageId:"+messageId);
            }
            //删除磁盘数据
            if(message.getDeliveryMode()==0){
                diskDataCenter.deleteMessageFromQueue(queue,message);
            }
            //删除未确认消息队列集合中的消息
            memoryDataCenter.deleteWaitAckQueue(queueName,messageId);
            //删除消息集合中的数据
            memoryDataCenter.deleteMessage(messageId);
            System.out.println("[VirtualHost] basicAck 消息被成功确认 queueName:"+queueName
            +" ,messageId:"+messageId);
            return true;
        } catch (MqException | IOException | ClassNotFoundException e) {
            System.out.println("[VirtualHost] basicAck 消息确认失败 queueName:"+queueName
            +" ,messageId:"+messageId);
            e.printStackTrace();
            return false;
        }
    }


}
